package com.rosense.nfvo.util;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import com.rosense.commons.io.IOUtils;
import com.rosense.commons.io.StreamUtils;
import com.rosense.sa.bean.Call;
import com.rosense.sa.bean.SaMessage;


public class CallMsgsHdfsQuery {

	public List<String> fNames = new ArrayList<String>();
	public List<Long> positions = new ArrayList<Long>();
	public long stime;
	public long etime;

	public String hdfsData = "/user/hive/hdfs_5gznyy/data/tracesa2/tracedata/";

	public FileSystem fs = null;

	public CallMsgsHdfsQuery(String fNames, Long positions,long stime,long etime) {
		try {
			this.fNames.add(fNames);
			this.positions.add(positions);
			this.stime=stime;
			this.etime=etime;
			this.fs = Kbs2HdfsUtil.getFileSystem();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public CallMsgsHdfsQuery(Call call) {
		try {
			this.fNames.addAll(call.fName);
			this.positions.addAll(call.position);
			this.stime=call.startTime;
			this.etime=call.endTime;
			this.fs = Kbs2HdfsUtil.getFileSystem();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public List<SaMessage> query() {
		List<SaMessage> readList=new ArrayList<SaMessage>();
		for (int i = 0; i < fNames.size(); i++) {
			String fName = fNames.get(i);
			long position = positions.get(i);
			String ymdhm=fName.split("_")[0];
			FSDataInputStream in = null;
			try {
				Path path = new Path(this.hdfsData+ymdhm+"/"+fName);
				System.out.println(path.toString() + "," + position);
				in = fs.open(path);
	            in.skip(position);
	            
	            int callLength=StreamUtils.readInt(in);
	            int msgLength=StreamUtils.readInt(in);
	            long amfuengapid=StreamUtils.readLong(in);
	            short amfid=StreamUtils.readShort(in);
	            ByteBuffer buffer = readByte(in, msgLength);
	            buffer.position(0);
	            while (buffer.hasRemaining()) {
	            	SaMessage msg=new SaMessage(buffer);
	            	if(msg.millisecond>=stime&&msg.millisecond<=etime) {
	            		readList.add(msg);
	            	}
	            }
			} catch (Exception e) {
				e.printStackTrace();
			} finally {
				IOUtils.closeIO(in);
			}
		}
		
		
		return readList;
	}
	
	 private static ByteBuffer readByte(FSDataInputStream open, int msglenth) {
	        ByteBuffer buffer = ByteBuffer.allocate(msglenth);
	        int len = msglenth;
	        while (len > 0) {
	            try {
	                byte[] bb2 = new byte[len];
	                int end = open.read(bb2);
	                len = len - end;
	                if (end > -1) {
	                    buffer.put(bb2, 0, end);
	                } else {
	                    return null;
	                }
	            } catch (Exception e) {
	                e.printStackTrace();
	                return null;
	            }
	        }
	        buffer.position(0);
	        return buffer;
	    }

	public void release() {
		if (fs != null)
			try {
				fs.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
	}

}
